﻿#include "stdafx.h"
#include "TcpClient.h"

MemPool<uv_connect_t> TcpClient::s_mp_uvConnect;
MemPool<uv_write_t> TcpClient::s_mp_uvWrite;
MemPool<uv_buf_t> TcpClient::s_mp_uvBuf;

TcpClient::TcpClient(size_t maxBufSize, uv_loop_t * loop)
{
    m_maxBufSize = maxBufSize;
    m_recvBuf = (char*)malloc(maxBufSize);
    assert(m_recvBuf);
    m_loop = loop;
    int r = uv_tcp_init(m_loop, &m_handle);
    m_handle.data = this;
}

TcpClient::~TcpClient()
{
    close();
    if (m_recvBuf) {
        free(m_recvBuf);
        m_recvBuf = nullptr;
    }
}

void TcpClient::connect(const char *host, u_short port)
{
    uv_getaddrinfo_t req_addr;
    addrinfo ai;
    char sPort[32];
    int status;

    if (strcmp(host, "::") == 0)
        host = "localhost";

    sprintf(sPort, "%d", port);
    ai.ai_family = PF_INET;
    ai.ai_socktype = SOCK_STREAM;
    ai.ai_protocol = IPPROTO_TCP;
    ai.ai_flags = 0;
    ai.ai_addrlen = sizeof(addrinfo);
    status = uv_getaddrinfo(m_loop, &req_addr, nullptr, host, sPort, &ai);
    if (status < 0) {
        auto err = uv_strerror(status);
        assert("uv_getaddrinfo err!");
        return;
    }
    auto req_conn = s_mp_uvConnect.alloc();
    req_conn->data = this;
    status = uv_tcp_connect(req_conn, &m_handle, req_addr.addrinfo->ai_addr, on_connect);
    uv_freeaddrinfo(req_addr.addrinfo);
    if (status < 0) {
        auto err = uv_strerror(status);
        assert("uv_tcp_connect err!");
    }
}

void TcpClient::close()
{
    if (isConnected()) {
        uv_close((uv_handle_t*)&m_handle, on_close);
    }
}

void TcpClient::beginRead()
{
    int r = uv_read_start((uv_stream_t*)&m_handle, &on_alloc, on_read);
    r = r;
}

void TcpClient::write(const void * data, size_t dataLen)
{
    auto req = s_mp_uvWrite.alloc();
    auto bufs = s_mp_uvBuf.alloc();
    req->data = bufs;
    bufs->base = new char[dataLen];
    bufs->len = dataLen;
    memcpy(bufs->base, data, dataLen);
    int r = uv_write(req, (uv_stream_t*)&m_handle, bufs, 1, on_write);
}

char *TcpClient::peek(size_t *retLen)
{
	*retLen = m_recvLen;
	return m_recvBuf + m_recvBuf_dataPos;
}

void TcpClient::fetch(size_t num)
{
    assert(num <= m_recvLen);
    m_recvBuf_dataPos += num;
    m_recvLen -= num;
    if (m_recvLen == 0) {
        m_recvBuf_dataPos = 0;
    }
    else if (m_recvBuf_dataPos > m_maxBufSize / 2) {
        // 如果数据已经保存到后半部分了，则移动到开头位置
        memcpy(m_recvBuf, m_recvBuf + m_recvBuf_dataPos, m_recvLen);
        m_recvBuf_dataPos = 0;
    }
}

void TcpClient::on_connect(uv_connect_t * req, int status)
{
    auto _this = (TcpClient*)req->data;
    s_mp_uvConnect.free(req);
    if (_this->m_listener) {
        _this->m_listener->on_connect(_this, status);
    }
    if (status < 0) {
        auto err = uv_strerror(status);
        assert("on_connect err!");
    }
}

void TcpClient::on_alloc(uv_handle_t * handle, size_t suggested_size, uv_buf_t * buf)
{
    auto _this = (TcpClient*)handle->data;
    size_t endPos = _this->m_recvBuf_dataPos + _this->m_recvLen;
    size_t needSize = endPos + suggested_size;
    if (needSize > _this->m_maxBufSize) {
        // 扩容，增加1/3大小，如果不够则使用needSize
        size_t newSize = _this->m_maxBufSize + _this->m_maxBufSize / 3;
        if (needSize > newSize) newSize = needSize;
        auto reAllocPtr = (char*)realloc(_this->m_recvBuf, newSize);
        assert(reAllocPtr);
        _this->m_recvBuf = reAllocPtr;
    }

    buf->base = _this->m_recvBuf + endPos;
    buf->len = suggested_size;
}

void TcpClient::on_read(uv_stream_t * stream, ssize_t nread, const uv_buf_t * buf)
{
    auto _this = (TcpClient*)stream->data;
    if (nread > 0) {
        _this->m_recvLen += nread;
        if (_this->m_listener) {
            size_t nget = _this->m_listener->on_read(_this, _this->m_recvBuf + _this->m_recvBuf_dataPos, _this->m_recvLen);
            if (nget > 0) {
                _this->fetch(nget);
            }
        }
    }
    else if (nread < 0) {
        _this->close();
    }
}

void TcpClient::on_write(uv_write_t * req, int status)
{
    auto buf = (uv_buf_t*)req->data;
    delete[] buf->base;
    buf->base = nullptr;
    buf->len = 0;
    s_mp_uvBuf.free(buf);
    s_mp_uvWrite.free(req);
}

void TcpClient::on_close(uv_handle_t * handle)
{
    auto _this = (TcpClient*)handle->data;
    if (_this->m_listener)
        _this->m_listener->on_close(_this);
}
